package com.ucode.springboot.starter.canal.client.transfer;

import java.lang.reflect.Method;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.function.Predicate;
import org.springframework.util.StringUtils;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.CanalEntry.EntryType;
import com.alibaba.otter.canal.protocol.CanalEntry.EventType;
import com.alibaba.otter.canal.protocol.CanalEntry.RowChange;
import com.ucode.springboot.starter.canal.annotation.ListenPoint;
import com.ucode.springboot.starter.canal.client.CanalMsg;
import com.ucode.springboot.starter.canal.client.ListenerPoint;
import com.ucode.springboot.starter.canal.config.CanalConfig.Instance;
import com.ucode.springboot.starter.canal.event.CanalEventListener;

/**
 * 默认的消息转换器
 * @author: liliang
 * @date: 2020年2月27日 下午9:09:52
 */
public class DefaultMessageTransponder extends AbstractBasicMessageTransponder{

    public DefaultMessageTransponder(CanalConnector connector, Entry<String, Instance> config, List<CanalEventListener> listeners,
            List<ListenerPoint> annoListeners) {
        super(connector, config, listeners, annoListeners);
    }

    @Override
    protected Predicate<Entry<Method, ListenPoint>> getAnnotationFilter(String destination, String schemaName, String tableName,
            EventType eventType) {
        //看看canal instances key是否正确
        Predicate<Map.Entry<Method, ListenPoint>> df = e -> StringUtils.isEmpty(e.getValue().destination())
                || e.getValue().destination().equals(destination) || destination == null;
        
        //看看数据库实例名是否一样
        Predicate<Map.Entry<Method, ListenPoint>> sf = e -> e.getValue().schema().length == 0
                || Arrays.stream(e.getValue().schema()).anyMatch(s -> s.equals(schemaName)) || schemaName == null;
        
        //看看表名是否一样
        Predicate<Map.Entry<Method, ListenPoint>> tf = e -> e.getValue().table().length == 0
                || Arrays.stream(e.getValue().table()).anyMatch(t -> t.equals(tableName)) || tableName == null;
        
        //类型一致
        Predicate<Map.Entry<Method, ListenPoint>> ef = e -> e.getValue().eventType().length == 0
                || Arrays.stream(e.getValue().eventType()).anyMatch(ev -> ev == eventType) || eventType == null;
        
        return df.and(sf).and(tf).and(ef);
    }

    @Override
    protected Object[] getInvokeArgs(Method method, CanalMsg canalMsg, RowChange rowChange) {
        Class<?>[] parameterTypes= method.getParameterTypes();
        Object[] invokeArgs = new Object[parameterTypes.length];
        for (int i = 0; i < parameterTypes.length; i++) {
            Object obj = null;
            if(parameterTypes[i] == CanalMsg.class){
                obj = canalMsg;
            }else if(parameterTypes[i] == CanalEntry.RowChange.class){
                obj = rowChange;
            }
            invokeArgs[i] = obj;
        }
        return invokeArgs;
    }

    @Override
    protected List<EntryType> getIgnoreEntryTypes() {
        // 事物数据不处理
        return Arrays.asList(CanalEntry.EntryType.TRANSACTIONBEGIN, CanalEntry.EntryType.TRANSACTIONEND, CanalEntry.EntryType.HEARTBEAT);
    }

}
